package com.openness.example.hbase;

import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.coprocessor.AggregateProtocol;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @author Rain
 * @version 2013-5-21 下午4:56:12
 */

public class RowCountMain {

	public static void main(String[] args) throws Throwable {
		Configuration conf = new Configuration();

		conf.set("hbase.master", "leaderman.rain.centos:60010");
		conf.set("hbase.zookeeper.quorum", "leaderman.rain.centos");
		conf.set("hbase.zookeeper.property.clientPort", "2181");

		HTable table = new HTable(conf, "student");

		Map<byte[], Long> result = table.coprocessorExec(
				AggregateProtocol.class, null, null,
				new Batch.Call<AggregateProtocol, Long>() {

					@Override
					public Long call(AggregateProtocol protocol)
							throws IOException {
						Scan scan = new Scan();
						scan.addFamily(Bytes.toBytes("baseinfo"));
						
						return protocol.getRowNum(new LongColumnInterpreter(),
								scan);
					}

				});

		long rowCount = 0;

		for (Entry<byte[], Long> entry : result.entrySet()) {
			rowCount += entry.getValue();
		}
		
		System.out.println(rowCount);

		table.close();
	}
}
